package com.wejoy.video.middle.station.data.event;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.Collections;

/**
 * 基础监听器
 */
public class BaseListener {
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    protected RedisTemplate redisTemplate;

    /**
     * 初始化流消息
     * @param redisTemplate
     * @param streamKey
     * @param groupId
     */
    protected void initStream(RedisTemplate redisTemplate, String streamKey, String groupId) {
        // 判断 stream 是否初始化，未初始化则进行初始化
        if (!Boolean.TRUE.equals(redisTemplate.hasKey(streamKey))) {
            // 往 stream 发送消息，会自动创建 stream
            RecordId recordId = redisTemplate.opsForStream().add(streamKey, Collections.singletonMap("_up", "up"));
            // 删除刚刚创建的消息
            redisTemplate.opsForStream().delete(streamKey, recordId);
        }
        // 创建 消费者组
        StreamInfo.XInfoGroups groups = redisTemplate.opsForStream().groups(streamKey);
        if (null != groups) {
            boolean noneMatch = groups.stream().noneMatch(item -> item.groupName().equals(groupId));
            if (noneMatch) {
                redisTemplate.opsForStream().createGroup(streamKey, groupId);
            }
        } else {
            redisTemplate.opsForStream().createGroup(streamKey, groupId);
        }
    }

    /**
     * 手动确认ack
     * @param groupId
     * @param message
     */
    protected void acknowledge(String groupId, MapRecord<String, String, String> message) {
        Long acknowledge = redisTemplate.opsForStream().acknowledge(groupId, message);
        logger.info("确认ack: {}", acknowledge);
    }
}
